package jrain.fw.mq.ons;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;

import jrain.fw.core.cfg.CfgService;
import jrain.fw.core.mq.MqConsumerService;
import jrain.fw.core.mq.SubscribeCallbak;
import jrain.fw.core.service.DestroyService;
import jrain.fw.core.service.InitService;
import jrain.fw.core.service.Service;
import jrain.fw.core.service.ServiceConst;
import jrain.fw.core.utils.FwUtils;
import jrain.utils.collection.RMap;
import jrain.utils.consts.BaseConst;
import jrain.utils.lang.LangUtils;
import jrain.utils.lang.StringUtils;

@Service(name = "ons", group = ServiceConst.S_MQ_CONSUMER_SERVICE, module = ServiceConst.M_MQ)
public class MqOnsConsumerService implements MqConsumerService, InitService, DestroyService {

	private static final AtomicBoolean init = new AtomicBoolean(false);
	private Properties properties = new Properties();
	private List<Consumer> consumers = new ArrayList<>();
	private List<OrderConsumer> orderConsumers = new ArrayList<>();

	public void subscribe(String topic, String tag, SubscribeCallbak callbak) {
		Consumer consumer = ONSFactory.createConsumer(properties);
		consumer.subscribe(topic, tag, new MessageListener() {
			@Override
			public Action consume(Message message, ConsumeContext context) {
				String data = new String(message.getBody(), BaseConst.UTF8);
				boolean b = callbak.handler(topic, tag, data, message.getKey(), message.getMsgID());
				if (b) {
					return Action.CommitMessage;
				} else {
					return Action.ReconsumeLater;
				}
			}
		});
		consumer.start();
		consumers.add(consumer);

	}

	public void subscribeOrder(String topic, String tag, SubscribeCallbak callbak) {
		OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);
		consumer.subscribe(topic, tag, new MessageOrderListener() {
			public OrderAction consume(Message message, ConsumeOrderContext context) {
				String data = new String(message.getBody(), BaseConst.UTF8);
				boolean b = callbak.handler(topic, tag, data, message.getKey(), message.getMsgID());
				if (b) {
					return OrderAction.Success;
				} else {
					return OrderAction.Suspend;
				}
			}
		});
		consumer.start();
		orderConsumers.add(consumer);
	}

	public void init() {
		if (!init.getAndSet(true)) {
			CfgService cfgService = FwUtils.getCfgService();
			RMap<String, Object> cfgMap = cfgService.getCfgMap(ServiceConst.M_MQ, null);
			if (cfgMap != null) {
				boolean enable = LangUtils.parseBoolean(cfgMap.get("enable"), false);
				if (enable) {
					Properties properties = new Properties();
					String producerId = StringUtils.trimNull(cfgMap.get("mq.producer.id"));
					String accessKey = StringUtils.trimNull(cfgMap.get("mq.access.key"));
					String secretKey = StringUtils.trimNull(cfgMap.get("mq.secret.key"));
					String onsAddr = StringUtils.trimNull(cfgMap.get("mq.ons.addr"));
					properties.put(PropertyKeyConst.ProducerId, producerId);
					// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
					properties.put(PropertyKeyConst.AccessKey, accessKey);
					// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
					properties.put(PropertyKeyConst.SecretKey, secretKey);
					// 设置 TCP 接入域名（此处以公共云生产环境为例）
					properties.put(PropertyKeyConst.ONSAddr, onsAddr);
				}
			}
		}
	}

	@Override
	public void destroy() {
		if (init.get()) {
			for (Consumer consumer : consumers) {
				try {
					consumer.shutdown();
				} catch (Throwable t) {
				}
			}
			for (OrderConsumer consumer : orderConsumers) {
				try {
					consumer.shutdown();
				} catch (Throwable t) {
				}
			}
			init.set(false);
		}
	}

}
